package Source

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.streaming.api.scala._
import java.util.Properties

object SourceFromKafka {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //配置项
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    //这边是要读取Kafka的数据，所以算是一个消费者
    /**
     * 第一个泛型：读取过来的数据类型
     * 第一个参数：Kafka名
     * 第二个参数：序列化
     * 第三个参数：配置项
     */
    val stream =
      env.addSource(new FlinkKafkaConsumer011[String]
      ("first", new SimpleStringSchema(), properties))

    stream.print()
    env.execute()
  }
}
